﻿#include "MdbTables.h"
#include "Mdb.h"
#include "Logger.h"
#include <string>
#include <string.h>
#include <vector>

using std::string;
using std::vector;

namespace mdb
{
	TickTable::TickTable(Mdb* mdb)
		:m_Mdb(mdb)
	{
		m_MdbSubscriber = nullptr;
		m_PrimaryKey = new TickPrimaryKey(this);
		m_InstrumentIDIndex = new TickIndexInstrumentID(this);
	}
	TickTable::~TickTable()
	{
		delete m_PrimaryKey;
		m_PrimaryKey = nullptr;
		delete m_InstrumentIDIndex;
		m_InstrumentIDIndex = nullptr;
	}
	void TickTable::Subscribe(MdbSubscriber* mdbSubscriber)
	{
		m_MdbSubscriber = mdbSubscriber;
	}
	void TickTable::UnSubscribe()
	{
		m_MdbSubscriber = nullptr;
	}
	void TickTable::LockShared()
	{
		m_SharedMutex.lock_shared();
	}
	void TickTable::UnlockShared()
	{
		m_SharedMutex.unlock_shared();
	}
	void TickTable::InitDB()
	{
		std::shared_lock guard(m_SharedMutex);
		for (auto it = m_PrimaryKey->m_Index.begin(); it != m_PrimaryKey->m_Index.end(); ++it)
		{
			m_MdbSubscriber->OnTickReplace(new Tick(**it));
		}
		m_DBInited = true;
	}
	bool TickTable::Insert(Tick* record)
	{
		std::lock_guard guard(m_SharedMutex);
		if (!m_PrimaryKey->CheckInsert(record))
		{
			WriteLog(LogLevel::Warning, "Insert Failed for Tick:[%s]", record->GetString());
			record->Free();
			return false;
		}

		m_PrimaryKey->Insert(record);

		m_InstrumentIDIndex->Insert(record);
		
		if (m_MdbSubscriber != nullptr && m_DBInited)
		{
			m_MdbSubscriber->OnTickInsert(record);
		}
		return true;
	}
	void TickTable::Erase(Tick* record)
	{
		std::lock_guard guard(m_SharedMutex);
		EraseUniqueKey(record);
		EraseIndex(record);
		if (m_MdbSubscriber != nullptr && m_DBInited)
		{
			m_MdbSubscriber->OnTickErase(record);
		}
		else
		{
			record->Free();
		}
	}
	int TickTable::EraseByInstrumentIDIndex(const InstrumentIDType& InstrumentID)
	{
		m_InstrumentIDIndex->FillCompareRecord(InstrumentID);
		vector<Tick*> records;
		std::lock_guard guard(m_SharedMutex);
		auto range = m_InstrumentIDIndex->m_Index.equal_range(&t_CompareTick);
		for (auto& it = range.first; it != range.second; ++it)
		{
			records.push_back(*it);
		}
		for (auto record : records)
		{
			EraseUniqueKey(record);
			EraseIndex(record);
			record->Free();
		}
		if (m_MdbSubscriber != nullptr && m_DBInited)
		{
			auto record = Tick::Allocate();
			memcpy(record, &t_CompareTick, sizeof(Tick));
			m_MdbSubscriber->OnTickEraseByInstrumentIDIndex(record);
		}
		return (int)records.size();
	}
	bool TickTable::Update(Tick* const oldRecord, Tick* const newRecord, bool updateDB)
	{
		std::lock_guard guard(m_SharedMutex);
		if (!m_PrimaryKey->CheckUpdate(oldRecord, newRecord))
		{
			WriteLog(LogLevel::Warning, "Update Failed for Tick:[%s]", oldRecord->GetString());
			WriteLog(LogLevel::Warning, "              New Tick:[%s]", newRecord->GetString());
			newRecord->Free();
			return false;
		}

		bool InstrumentIDIndexUpdate = m_InstrumentIDIndex->NeedUpdate(oldRecord, newRecord);
		TickIndexInstrumentID::iterator itInstrumentID;
		if (InstrumentIDIndexUpdate)
		{
			itInstrumentID = m_InstrumentIDIndex->FindNode(oldRecord);
		}
		::memcpy((void*)oldRecord, newRecord, sizeof(Tick));
		if (InstrumentIDIndexUpdate)
		{
			m_InstrumentIDIndex->Update(itInstrumentID);
		}

		if (updateDB && m_MdbSubscriber != nullptr && m_DBInited)
		{
			m_MdbSubscriber->OnTickUpdate(newRecord);
		}
		else
		{
			newRecord->Free();
		}
		return true;
	}
	void TickTable::TruncateTable()
	{
		std::lock_guard guard(m_SharedMutex);
		for (auto it = m_PrimaryKey->m_Index.begin(); it != m_PrimaryKey->m_Index.end(); ++it)
		{
			(*it)->Free();
		}
		m_PrimaryKey->m_Index.clear();
		m_InstrumentIDIndex->m_Index.clear();
		if (m_MdbSubscriber != nullptr && m_DBInited)
		{
			m_MdbSubscriber->OnTickTruncate();
		}
	}
	void TickTable::Dump(const char* dir)
	{
		string fileName = string(dir) + "//t_Tick.csv";
		FILE* dumpFile = fopen(fileName.c_str(), "w");
		if (dumpFile == nullptr)
		{
			return;
		}

		fprintf(dumpFile, "TradingDay,ExchangeID,InstrumentID,UpdateTs,LastPrice,LastTraded,Volume,OpenInterest,Turnover,BidPrice1,BidVolume1,AskPrice1,AskVolume1,PreSettlementPrice,PreClosePrice,PreOpenInterest,HighestPrice,LowestPrice\n");
		char buff[4096] = { 0 };
		std::shared_lock guard(m_SharedMutex);
		for (auto it = m_PrimaryKey->m_Index.begin(); it != m_PrimaryKey->m_Index.end(); ++it)
		{
			fprintf(dumpFile, "%s\n", (*it)->GetString());
		}
		fclose(dumpFile);
	}
	void TickTable::EraseUniqueKey(Tick* record)
	{
		m_PrimaryKey->Erase(record);
	}
	void TickTable::EraseIndex(Tick* record)
	{
		m_InstrumentIDIndex->Erase(record);
	}

	BarTable::BarTable(Mdb* mdb)
		:m_Mdb(mdb)
	{
		m_MdbSubscriber = nullptr;
		m_PrimaryKey = new BarPrimaryKey(this);
		m_InstrumentIDIndex = new BarIndexInstrumentID(this);
		m_TradingDayIndex = new BarIndexTradingDay(this);
	}
	BarTable::~BarTable()
	{
		delete m_PrimaryKey;
		m_PrimaryKey = nullptr;
		delete m_InstrumentIDIndex;
		m_InstrumentIDIndex = nullptr;
		delete m_TradingDayIndex;
		m_TradingDayIndex = nullptr;
	}
	void BarTable::Subscribe(MdbSubscriber* mdbSubscriber)
	{
		m_MdbSubscriber = mdbSubscriber;
	}
	void BarTable::UnSubscribe()
	{
		m_MdbSubscriber = nullptr;
	}
	void BarTable::LockShared()
	{
		m_SharedMutex.lock_shared();
	}
	void BarTable::UnlockShared()
	{
		m_SharedMutex.unlock_shared();
	}
	void BarTable::InitDB()
	{
		std::shared_lock guard(m_SharedMutex);
		for (auto it = m_PrimaryKey->m_Index.begin(); it != m_PrimaryKey->m_Index.end(); ++it)
		{
			m_MdbSubscriber->OnBarReplace(new Bar(**it));
		}
		m_DBInited = true;
	}
	bool BarTable::Insert(Bar* record)
	{
		std::lock_guard guard(m_SharedMutex);
		if (!m_PrimaryKey->CheckInsert(record))
		{
			WriteLog(LogLevel::Warning, "Insert Failed for Bar:[%s]", record->GetString());
			record->Free();
			return false;
		}

		m_PrimaryKey->Insert(record);

		m_InstrumentIDIndex->Insert(record);
		m_TradingDayIndex->Insert(record);
		
		if (m_MdbSubscriber != nullptr && m_DBInited)
		{
			m_MdbSubscriber->OnBarInsert(record);
		}
		return true;
	}
	void BarTable::Erase(Bar* record)
	{
		std::lock_guard guard(m_SharedMutex);
		EraseUniqueKey(record);
		EraseIndex(record);
		if (m_MdbSubscriber != nullptr && m_DBInited)
		{
			m_MdbSubscriber->OnBarErase(record);
		}
		else
		{
			record->Free();
		}
	}
	int BarTable::EraseByInstrumentIDIndex(const Int64Type& TradingDay, const ExchangeIDType& ExchangeID, const InstrumentIDType& InstrumentID)
	{
		m_InstrumentIDIndex->FillCompareRecord(TradingDay, ExchangeID, InstrumentID);
		vector<Bar*> records;
		std::lock_guard guard(m_SharedMutex);
		auto range = m_InstrumentIDIndex->m_Index.equal_range(&t_CompareBar);
		for (auto& it = range.first; it != range.second; ++it)
		{
			records.push_back(*it);
		}
		for (auto record : records)
		{
			EraseUniqueKey(record);
			EraseIndex(record);
			record->Free();
		}
		if (m_MdbSubscriber != nullptr && m_DBInited)
		{
			auto record = Bar::Allocate();
			memcpy(record, &t_CompareBar, sizeof(Bar));
			m_MdbSubscriber->OnBarEraseByInstrumentIDIndex(record);
		}
		return (int)records.size();
	}
	int BarTable::EraseByTradingDayIndex(const Int64Type& TradingDay)
	{
		m_TradingDayIndex->FillCompareRecord(TradingDay);
		vector<Bar*> records;
		std::lock_guard guard(m_SharedMutex);
		auto range = m_TradingDayIndex->m_Index.equal_range(&t_CompareBar);
		for (auto& it = range.first; it != range.second; ++it)
		{
			records.push_back(*it);
		}
		for (auto record : records)
		{
			EraseUniqueKey(record);
			EraseIndex(record);
			record->Free();
		}
		if (m_MdbSubscriber != nullptr && m_DBInited)
		{
			auto record = Bar::Allocate();
			memcpy(record, &t_CompareBar, sizeof(Bar));
			m_MdbSubscriber->OnBarEraseByTradingDayIndex(record);
		}
		return (int)records.size();
	}
	bool BarTable::Update(Bar* const oldRecord, Bar* const newRecord, bool updateDB)
	{
		std::lock_guard guard(m_SharedMutex);
		if (!m_PrimaryKey->CheckUpdate(oldRecord, newRecord))
		{
			WriteLog(LogLevel::Warning, "Update Failed for Bar:[%s]", oldRecord->GetString());
			WriteLog(LogLevel::Warning, "              New Bar:[%s]", newRecord->GetString());
			newRecord->Free();
			return false;
		}

		bool InstrumentIDIndexUpdate = m_InstrumentIDIndex->NeedUpdate(oldRecord, newRecord);
		BarIndexInstrumentID::iterator itInstrumentID;
		if (InstrumentIDIndexUpdate)
		{
			itInstrumentID = m_InstrumentIDIndex->FindNode(oldRecord);
		}
		bool TradingDayIndexUpdate = m_TradingDayIndex->NeedUpdate(oldRecord, newRecord);
		BarIndexTradingDay::iterator itTradingDay;
		if (TradingDayIndexUpdate)
		{
			itTradingDay = m_TradingDayIndex->FindNode(oldRecord);
		}
		::memcpy((void*)oldRecord, newRecord, sizeof(Bar));
		if (InstrumentIDIndexUpdate)
		{
			m_InstrumentIDIndex->Update(itInstrumentID);
		}
		if (TradingDayIndexUpdate)
		{
			m_TradingDayIndex->Update(itTradingDay);
		}

		if (updateDB && m_MdbSubscriber != nullptr && m_DBInited)
		{
			m_MdbSubscriber->OnBarUpdate(newRecord);
		}
		else
		{
			newRecord->Free();
		}
		return true;
	}
	void BarTable::TruncateTable()
	{
		std::lock_guard guard(m_SharedMutex);
		for (auto it = m_PrimaryKey->m_Index.begin(); it != m_PrimaryKey->m_Index.end(); ++it)
		{
			(*it)->Free();
		}
		m_PrimaryKey->m_Index.clear();
		m_InstrumentIDIndex->m_Index.clear();
		m_TradingDayIndex->m_Index.clear();
		if (m_MdbSubscriber != nullptr && m_DBInited)
		{
			m_MdbSubscriber->OnBarTruncate();
		}
	}
	void BarTable::Dump(const char* dir)
	{
		string fileName = string(dir) + "//t_Bar.csv";
		FILE* dumpFile = fopen(fileName.c_str(), "w");
		if (dumpFile == nullptr)
		{
			return;
		}

		fprintf(dumpFile, "TradingDay,ExchangeID,InstrumentID,UpdateTs,Prece,LastTraded,Volume,Turnover,OpenInterest,Open,High,Low,Close,HighestPrice,LowestPrice,ProductClass,UnderlyingID\n");
		char buff[4096] = { 0 };
		std::shared_lock guard(m_SharedMutex);
		for (auto it = m_PrimaryKey->m_Index.begin(); it != m_PrimaryKey->m_Index.end(); ++it)
		{
			fprintf(dumpFile, "%s\n", (*it)->GetString());
		}
		fclose(dumpFile);
	}
	void BarTable::EraseUniqueKey(Bar* record)
	{
		m_PrimaryKey->Erase(record);
	}
	void BarTable::EraseIndex(Bar* record)
	{
		m_InstrumentIDIndex->Erase(record);
		m_TradingDayIndex->Erase(record);
	}

}

